var PACKET_SIZE = 64e3;
var MAX_ACK_PENDING = 3;

var FileFastOutputStream = function (client, file) {
	this.client = client;
	this.file = file;
}

FileFastOutputStream.prototype = {
	append: function (size) {
		if (this.closed) return;
		
		while (size > 0) {
			var to_write = size > this.file.blocksize ? this.file.blocksize : size;
			try {
				var block = this.client.host.rpc(
					this.client.nnAddress,
					this.client.nnipc,
					Namenode.newBlock,
					[this.client, this.file]);
			} catch (e) {
				throw e;
			}
			block.used = to_write;
			var numDatanodes = block.datanodes.length;
			for (var i = 0; i < numDatanodes; i++) {
				var datanode = this.client.host.network.resolve(block.datanodes[i]);
				if (!datanode) {
					throw "Error";
				}
				datanode.disk.create(block.id);
				datanode.disk.getInode(block.id).size = to_write;
				block.replicated ++;
			}
			
			size -= to_write;
		}	
	},
	
	close: function () {
		this.client.host.rpc(
			this.client.nnAddress,
			this.client.nnipc,
			Namenode.closeFile,
			this.file);
		console.log("file written at " + this.client.time());
	},
	
	flush: function () {
		
	}
}

/********************************************************/
	
var FileOutputStream = function (client, file) {
	this.client = client;
	this.file = file;
	this.partial = 0;
	this.offset_current = 0;
	this.offset_end = 0;
	this.block_offset = 0;
	this.streamer_running = false;
	this.current_block = null;
	this.acks_pending = 0;
	this.blockdone = new Sim.Event();
	this.ackdone = new Sim.Event();
	this.ackQueue = [];
	this.packetid = 0;
	
	this.socket = this.client.host.socket(this, this.recvack);
};

FileOutputStream.prototype = {
	append: function (size) {
		if (this.closed) return;
		
		this.partial += size;
		while (this.partial >= PACKET_SIZE) {
			this.offset_end += PACKET_SIZE;
			this.partial -= PACKET_SIZE;
		}
		if (this.offset_end > 0 && !this.sending) {
			this.streamer();
		}
	},
	
	flush: function () {
		this.offset_end += this.partial;
		this.partial = 0;
		if (this.offset_end > 0 && !this.sending) {
			this.streamer();
		}
	},
	
	close: function () {
		this.flush();
		this.closed = true;
	},
	
	streamer: function () {
		this.client.log("Streamer offset current=" + this.offset_current
				+ " end = " + this.offset_end);
		this.sending = true;
				
		if (this.offset_current > this.offset_end) {
			throw new Error("offset_current is ahead of offset_end?");
		}

		if (this.offset_current == this.offset_end) {
			// Nothing more to send for now
			
			// see if the session is closed.. in which case we will send a 0 size packet
			if (this.closed) {
				if (this.current_block) {
					this.socket.send({close: true, 
										block: this.current_block, 
										size: 16}, 16);
				}
				this.client.host.rpc(
					this.client.nnAddress,
					this.client.nnipc,
					Namenode.closeFile,
					this.file);
				this.client.host.closesocket(this.socket.port);
				console.log("file written at " + this.client.time());
			}
			this.sending = false;
			return;
		}
		
		if (!this.current_block) {
			try {
				this.current_block = this.client.host.rpc(
					this.client.nnAddress,
					this.client.nnipc,
					Namenode.newBlock,
					[this.client, this.file]);
			} catch (e) {
				if (e === Constants.HDFS.PREV_BLOCK_BUSY) {
					this.client.setTimer(1).done(this.streamer, this);
					return;
				}
				throw e;
			}
			
			this.block_offset = 0;
			this.socket.connect(this.current_block.datanodes[0], 
					HadoopConf.getValue("dfs.datanode.address"));
		}
		
		this.client.waitEvent(this.blockdone).done(this.streamer, this);
		
		this.blockstreamer();
	},

	blockstreamer: function () {
		var to_send = Math.min(
				this.offset_end - this.offset_current,
				this.file.blocksize - this.block_offset);
		
		this.client.log("BlockStreamer to_send=" + to_send
				+ " acks_pending=" + this.acks_pending
				+ " Packet size = " + PACKET_SIZE);
		
		// If there is something to send
		while (to_send > 0) {
			if (this.acks_pending == MAX_ACK_PENDING) {
				this.client.waitEvent(this.ackdone).done(this.blockstreamer, this);
				return;
			}
			
			// send the packet now
			var packet = {
				id: this.packetid ++,
				block: this.current_block,
				src: this.client.host.ip,
				sport: this.socket.port,
				size: to_send >= PACKET_SIZE ? PACKET_SIZE : to_send,
				acked: 0
			};
			
			this.block_offset += packet.size;
			this.offset_current += packet.size;
			to_send -= packet.size;
			this.acks_pending ++;
			this.ackQueue.push(packet);
			
			this.client.log("Sending packet " + packet.id + " of size " + packet.size);
			this.socket.send(packet, packet.size);
			
			to_send = Math.min(
					this.offset_end - this.offset_current,
					this.file.blocksize - this.block_offset);
		}
		
		// We don't have anything else to send.
		// Wait till we get response for all acks
		if (this.acks_pending > 0) {
			this.client.waitEvent(this.ackdone).done(this.blockstreamer, this);
			return;
		}
		
		// now we are completely done. Time to wake up the streamer
		// first check if we have completed the current block
		if (this.file.blocksize == this.block_offset) {
			this.current_block = null;
		}
		this.blockdone.fire();
	},
	
	recvack: function (src, sport, dport, packet, size) {
		// TODO: we are not checking if this is a duplicate ack
		packet.acked ++;
		var releaseBlockStreamer = false;
		while (true) {
			if (this.ackQueue.length == 0) break;
			packet = this.ackQueue[0];
			if (packet.acked < packet.block.file.replication) break;
			this.ackQueue.shift();
			this.acks_pending --;
			this.current_block.size += packet.size;
			this.file.size += packet.size;
			releaseBlockStreamer = true;
		}
		
		if (releaseBlockStreamer) {
			this.ackdone.fire();
		}
	}
};
/***********************************************************************/
var FileInputStream = function (client, file) {
	this.client = client;
	this.file = file;
	this.done = new Sim.Event();
	this.socket = this.client.host.socket(this, this.recv);
};

FileInputStream.prototype = {
	read: function (start, end) {
		if (this.reading) {
			throw Constants.FS.FILE_BUSY;
		}
		
		this.start = start ? start : 0;
		this.end = end ? end: this.file.size;
		
		if (this.end > this.file.size) {
			this.end = this.file.size;
		}
		
		if (this.start >= this.end) {
			throw Constants.FS.ILLEGAL_SEEK;
		}
		
		this.reading = true;
		this.readNextBlock();
		return this.client.waitEvent(this.done);
	},
	
	close: function () {
		
	},
	
	readNextBlock: function () {
		// find the block that must be read
		var blockidx = Math.floor(this.start / this.file.blocksize);
		var block = this.file.blocks[blockidx];
		var prev = blockidx * this.file.blocksize;
		
		// find block_from and length
		var block_from = this.start - prev;
		var length = Math.min(this.end - this.start, this.file.blocksize - block_from);
		this.block_end = this.start + length;
	
		// find the datanode
		var datanodeip = block.datanodes[0];
		
		// send message to datanode to send the block
		try {
			this.client.host.rpc(datanodeip,
								HadoopConf.getValue("dfs.datanode.ipc.address"),
								Datanode.getBlock,
								[this.client.host.ip, this.socket.port, block, block_from, length]);
		} catch (e) {
			// RPC failure
			// TODO
		}
		
		// the datanode will be sending us the data now
	},
	
	recv: function (src, sport, dport, block, length) {
		// we got data from datanode
		this.client.log("got data length = " + length);
		
		// increment this.start
		this.start += length;
		
		// we have read everything
		if (this.start == this.end) {
			this.reading = false;
			this.done.fire();
			return;
		}
		
		// we are done reading this block.. but there is more data to be read
		if (this.start == this.block_end) {
			this.readNextBlock();
		}
	}
};


/***********************************************************************/

var HdfsClient = {
	start: function (host) {
		this.host = host;
		this.name = "client";
		this.setTimer(0).done(this.begin);
	},
	begin: function () {
		this.nnAddress = HadoopConf.getValue('dfs.namenode.address');
		this.nnipc = HadoopConf.getValue('dfs.namenode.ipc.port');
	},
	createDir: function (fullpath) {
		
	},
	
	openFile: function (fullpath, mode, blocksize, replication, fast) {
		var file = this.host.rpc(
				this.nnAddress,
				this.nnipc,
				Namenode.openFile,
				[this, fullpath, mode, blocksize, replication]);
		if (mode == "r") {
			return new FileInputStream(this, file);
		} else if (mode == "w" || mode == "a") {
			if (fast) return new FileFastOutputStream(this, file);
			else return new FileOutputStream(this, file);
		}
	}
};